package cloudy.bolt;

import cloudy.hbase.dao.HBaseDAO;
import cloudy.hbase.dao.impl.HBaseDAOImpl;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.IBasicBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;

/**
 * @Name cloudy.bolt.AreaResBolt
 * @Description
 * @Author Elwyn
 * @Version 2017/9/29
 * @Copyright 上海云辰信息科技有限公司
 **/
public class AreaResBolt implements IBasicBolt {
	private Map<String, Double> countMap;
	long beginTime = System.currentTimeMillis();
	long endTime;
	HBaseDAO hBaseDAO;

	@Override
	public void prepare(Map map, TopologyContext topologyContext) {
		countMap = new HashMap<>();
		hBaseDAO = new HBaseDAOImpl();
	}

	@Override
	public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) {
		String date_area = tuple.getString(0);
		Double order_amt = tuple.getDouble(1);
		countMap.put(date_area, order_amt);
		endTime = System.currentTimeMillis();
		if (endTime - beginTime >= 5 * 1000) {
			for (String key : countMap.keySet()) {
				hBaseDAO.insert("area_order", key, "cf", "order_amt", String.valueOf(countMap.get(key)));
				System.err.println("resultBolt put hbase key =" + key + "order_amt : " + String.valueOf(countMap.get(key)));
			}
			//put into hbase
			beginTime = System.currentTimeMillis();
		}
	}

	@Override
	public void cleanup() {

	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		return null;
	}
}
